package sparkml_study

import org.apache.spark.sql.SparkSession
import org.apache.spark.ml.feature.StringIndexer
import org.apache.spark.ml.feature.IndexToString
import org.apache.spark.ml.feature.VectorIndexer
import org.apache.spark.ml.linalg.{Vector,Vectors}
object TeZhengZhuanHuan {
  def main(args: Array[String]): Unit = {
    /*
      1.StringIndexer:该转化器是把一列类别型的特征(或标签)进行编码，使其数值化
      ，索引的范围从0开始，该过程可以使得相应的特征索引化，使得某些无法接受类别型特征的算法可以使用
      ，并提高诸如决策树等机器学习算法的效率,索引构建的顺序为标签的频率，优先编码频率较大的标签，所以出现频率最高的标签为
      0号，如果输入的是数值型的，我们会把它转化为字符型，然后对其进行编码

      2.indexToString:它的作用与StringIndexer相反，它的作用是把标签索引的一列重新映射回原有的字符型标签
      ，然后一般和上面的那个联合使用，先使用上面的将字符串变成索引，然后在预测标签的时候又将索引变回字符串

     */

    val spark=SparkSession.builder()
      .master("local[*]")
      .appName("特征转换")
      .getOrCreate()

    import spark.implicits._

    //  创建实验数据
    val df1=spark.createDataFrame(Seq(
      (0, "a"),
      (1, "b"),
      (2, "c"),
      (3, "a"),
      (4, "a"),
      (5, "c")
    )).toDF("id", "category")

    //  创建stringindexer对象
    val indexer=new StringIndexer()
      .setInputCol("category")
      .setOutputCol("categoryindex")

    //  通过fit方法进行模型训练，然后对数据集进行处理
    val indexed1=indexer.fit(df1).transform(df1)
    //  查看结果
    indexed1.show(false)

    //  创建indexToString的转换器
    val idx2str: IndexToString = new IndexToString()
      .setInputCol("categoryindex")
      .setOutputCol("originalCategory")

    //  由于IndexToString是一个转化器，所以不需要fit可以直接训练
    //  它使用 StringIndexer 已经训练好的索引映射来反向转换数据，不需要再次训练。
    val indexString=idx2str.transform(indexed1)
    indexString.select("id","originalCategory").show(false)


    /*
        VectorIndexer:StringIndexer对单个类别型特征进行转换，如果这些特征都已经被组织再一个向量中
        ，又想对其中某些单个分量进行处理，则可以利用VectorIndexer类进行转换

        VectorIndexer类的maxCategories参数，可以自动识别类别型特征，并将原始值转化为类别索引，它基于特征值
        的数量来识别需要被类别化的特征，那些取值数最多不超过maxCategories的特征，将会被类型化并转化为索引
     */

    //  实验数据
    val data=Seq(
      Vectors.dense(-1.0,1.0,1.0),
      Vectors.dense(-1.0,3.0,1.0),
      Vectors.dense(0.0,5.0,1.0)
    )
    val df2=spark.createDataFrame(data.map(Tuple1.apply)).toDF("features")

    //  构建VectorIndexer转换器,设置输入和输出列，并进行模型训练
    //  这里设置setMaxCategories(2),即只有种类小于2的特征才被认为是类别型特征，否则被认为是连续型特征
    val indexerModel=new VectorIndexer()
      .setInputCol("features")
      .setOutputCol("indexed")
      .setMaxCategories(2)
      .fit(df2)

    //  接下来通过VectorIndexerModel的categoryMaps成员来获得被转换的特征及其映射，这里可以看到，共有两个特征被转换
    //  分别是0和2号
    val categoricalFeatures:Set[Int]=indexerModel.categoryMaps.keys.toSet

    println(s"Chose ${categoricalFeatures.size} categorical features:"+categoricalFeatures.mkString(","))

    //  使用转换器对数据进行训练
    val indexed=indexerModel.transform(df2)


    /*
       从结果可以看出来,只有种类小于2的特征才被认为是类别型特征，否则被认为是连续型特征。第0列和第2列的特征由于种类数量
       不超过2，被划分成类别型特征，并进行了索引，而而1(索引)列特征有三个值，因此不进行类型化的索引
     */
    indexed.show()




    spark.close()
  }

}
